diff --git a/common/djangoapps/student/views.py b/common/djangoapps/student/views.py
index ebd067942e..b1b8f6ff1e 100644
--- a/common/djangoapps/student/views.py
+++ b/common/djangoapps/student/views.py
@@ -1,3 +1,6 @@
+"""
+Student Views
+"""
 import datetime
 import feedparser
 import json
@@ -271,7 +274,7 @@ def dashboard(request):
             log.error("User {0} enrolled in non-existent course {1}"
                       .format(user.username, enrollment.course_id))
 
-    course_optouts = Optout.objects.filter(email=user.email).values_list('course_id', flat=True)
+    course_optouts = Optout.objects.filter(user=user).values_list('course_id', flat=True)
 
     message = ""
     if not user.is_active:
@@ -1289,13 +1292,13 @@ def change_email_settings(request):
     course_id = request.POST.get("course_id")
     receive_emails = request.POST.get("receive_emails")
     if receive_emails:
-        optout_object = Optout.objects.filter(email=user.email, course_id=course_id)
+        optout_object = Optout.objects.filter(user=user, course_id=course_id)
         if optout_object:
             optout_object.delete()
-        log.info(u"User {0} ({1}) opted to receive emails from course {2}".format(user.username, user.email, course_id))
+        log.info(u"User {0} ({1}) opted in to receive emails from course {2}".format(user.username, user.email, course_id))
         track.views.server_track(request, "change-email-settings", {"receive_emails": "yes", "course": course_id}, page='dashboard')
     else:
-        Optout.objects.get_or_create(email=request.user.email, course_id=course_id)
+        Optout.objects.get_or_create(user=user, course_id=course_id)
         log.info(u"User {0} ({1}) opted out of receiving emails from course {2}".format(user.username, user.email, course_id))
         track.views.server_track(request, "change-email-settings", {"receive_emails": "no", "course": course_id}, page='dashboard')
 
diff --git a/lms/djangoapps/bulk_email/admin.py b/lms/djangoapps/bulk_email/admin.py
index 40da36629c..3b40290f5d 100644
--- a/lms/djangoapps/bulk_email/admin.py
+++ b/lms/djangoapps/bulk_email/admin.py
@@ -13,7 +13,7 @@ class CourseEmailAdmin(admin.ModelAdmin):
 
 class OptoutAdmin(admin.ModelAdmin):
     """Admin for optouts."""
-    list_display = ('email', 'course_id')
+    list_display = ('user', 'course_id')
 
 
 admin.site.register(CourseEmail, CourseEmailAdmin)
diff --git a/lms/djangoapps/bulk_email/migrations/0001_initial.py b/lms/djangoapps/bulk_email/migrations/0001_initial.py
index 99c99d4efc..c3672a6de8 100644
--- a/lms/djangoapps/bulk_email/migrations/0001_initial.py
+++ b/lms/djangoapps/bulk_email/migrations/0001_initial.py
@@ -1,8 +1,6 @@
 # -*- coding: utf-8 -*-
-import datetime
 from south.db import db
 from south.v2 import SchemaMigration
-from django.db import models
 
 
 class Migration(SchemaMigration):
diff --git a/lms/djangoapps/bulk_email/migrations/0002_change_field_names.py b/lms/djangoapps/bulk_email/migrations/0002_change_field_names.py
index 95c0db339f..93fa33a544 100644
--- a/lms/djangoapps/bulk_email/migrations/0002_change_field_names.py
+++ b/lms/djangoapps/bulk_email/migrations/0002_change_field_names.py
@@ -1,8 +1,6 @@
 # -*- coding: utf-8 -*-
-import datetime
 from south.db import db
 from south.v2 import SchemaMigration
-from django.db import models
 
 
 class Migration(SchemaMigration):
@@ -19,7 +17,6 @@ class Migration(SchemaMigration):
                       self.gf('django.db.models.fields.TextField')(null=True, blank=True),
                       keep_default=False)
 
-
     def backwards(self, orm):
         # Renaming field 'CourseEmail.to_option'
         db.rename_column('bulk_email_courseemail', 'to_option', 'to')
@@ -30,9 +27,6 @@ class Migration(SchemaMigration):
         # Deleting field 'CourseEmail.text_message'
         db.delete_column('bulk_email_courseemail', 'text_message')
 
-
-
-
     models = {
         'auth.group': {
             'Meta': {'object_name': 'Group'},
diff --git a/lms/djangoapps/bulk_email/migrations/0003_add_optout_user.py b/lms/djangoapps/bulk_email/migrations/0003_add_optout_user.py
new file mode 100644
index 0000000000..1bf344f6e9
--- /dev/null
+++ b/lms/djangoapps/bulk_email/migrations/0003_add_optout_user.py
@@ -0,0 +1,91 @@
+# -*- coding: utf-8 -*-
+from south.db import db
+from south.v2 import SchemaMigration
+
+
+class Migration(SchemaMigration):
+
+    def forwards(self, orm):
+
+        # Adding field 'Optout.user'
+        db.add_column('bulk_email_optout', 'user',
+                      self.gf('django.db.models.fields.related.ForeignKey')(to=orm['auth.User'], null=True),
+                      keep_default=False)
+
+        # Removing unique constraint on 'Optout', fields ['course_id', 'email']
+        db.delete_unique('bulk_email_optout', ['course_id', 'email'])
+
+        # Adding unique constraint on 'Optout', fields ['course_id', 'user']
+        db.create_unique('bulk_email_optout', ['course_id', 'user_id'])
+
+    def backwards(self, orm):
+
+        # Removing unique constraint on 'Optout', fields ['course_id', 'user']
+        db.delete_unique('bulk_email_optout', ['course_id', 'user_id'])
+
+        # Deleting field 'Optout.email'
+        db.delete_column('bulk_email_optout', 'user_id')
+
+        # Creating unique constraint on 'Optout', fields ['course_id', 'email']
+        db.create_unique('bulk_email_optout', ['course_id', 'email'])
+
+    models = {
+        'auth.group': {
+            'Meta': {'object_name': 'Group'},
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'name': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '80'}),
+            'permissions': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Permission']", 'symmetrical': 'False', 'blank': 'True'})
+        },
+        'auth.permission': {
+            'Meta': {'ordering': "('content_type__app_label', 'content_type__model', 'codename')", 'unique_together': "(('content_type', 'codename'),)", 'object_name': 'Permission'},
+            'codename': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
+            'content_type': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['contenttypes.ContentType']"}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'name': ('django.db.models.fields.CharField', [], {'max_length': '50'})
+        },
+        'auth.user': {
+            'Meta': {'object_name': 'User'},
+            'date_joined': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime.now'}),
+            'email': ('django.db.models.fields.EmailField', [], {'max_length': '75', 'blank': 'True'}),
+            'first_name': ('django.db.models.fields.CharField', [], {'max_length': '30', 'blank': 'True'}),
+            'groups': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Group']", 'symmetrical': 'False', 'blank': 'True'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'is_active': ('django.db.models.fields.BooleanField', [], {'default': 'True'}),
+            'is_staff': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
+            'is_superuser': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
+            'last_login': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime.now'}),
+            'last_name': ('django.db.models.fields.CharField', [], {'max_length': '30', 'blank': 'True'}),
+            'password': ('django.db.models.fields.CharField', [], {'max_length': '128'}),
+            'user_permissions': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Permission']", 'symmetrical': 'False', 'blank': 'True'}),
+            'username': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '30'})
+        },
+        'bulk_email.courseemail': {
+            'Meta': {'object_name': 'CourseEmail'},
+            'course_id': ('django.db.models.fields.CharField', [], {'max_length': '255', 'db_index': 'True'}),
+            'created': ('django.db.models.fields.DateTimeField', [], {'auto_now_add': 'True', 'blank': 'True'}),
+            'html_message': ('django.db.models.fields.TextField', [], {'null': 'True', 'blank': 'True'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'modified': ('django.db.models.fields.DateTimeField', [], {'auto_now': 'True', 'blank': 'True'}),
+            'sender': ('django.db.models.fields.related.ForeignKey', [], {'default': '1', 'to': "orm['auth.User']", 'null': 'True', 'blank': 'True'}),
+            'slug': ('django.db.models.fields.CharField', [], {'max_length': '128', 'db_index': 'True'}),
+            'subject': ('django.db.models.fields.CharField', [], {'max_length': '128', 'blank': 'True'}),
+            'text_message': ('django.db.models.fields.TextField', [], {'null': 'True', 'blank': 'True'}),
+            'to_option': ('django.db.models.fields.CharField', [], {'default': "'myself'", 'max_length': '64'})
+        },
+        'bulk_email.optout': {
+            'Meta': {'unique_together': "(('user', 'course_id'),)", 'object_name': 'Optout'},
+            'course_id': ('django.db.models.fields.CharField', [], {'max_length': '255', 'db_index': 'True'}),
+            'email': ('django.db.models.fields.CharField', [], {'max_length': '255', 'db_index': 'True'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'user': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['auth.User']", 'null': 'True'})
+        },
+        'contenttypes.contenttype': {
+            'Meta': {'ordering': "('name',)", 'unique_together': "(('app_label', 'model'),)", 'object_name': 'ContentType', 'db_table': "'django_content_type'"},
+            'app_label': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'model': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
+            'name': ('django.db.models.fields.CharField', [], {'max_length': '100'})
+        }
+    }
+
+    complete_apps = ['bulk_email']
diff --git a/lms/djangoapps/bulk_email/migrations/0004_migrate_optout_user.py b/lms/djangoapps/bulk_email/migrations/0004_migrate_optout_user.py
new file mode 100644
index 0000000000..6dd2129466
--- /dev/null
+++ b/lms/djangoapps/bulk_email/migrations/0004_migrate_optout_user.py
@@ -0,0 +1,91 @@
+# -*- coding: utf-8 -*-
+from south.db import db
+from south.v2 import DataMigration
+from django.core.exceptions import ObjectDoesNotExist
+
+
+class Migration(DataMigration):
+
+    def forwards(self, orm):
+
+        # forwards data migration to copy over existing emails to associated ids
+        if not db.dry_run:
+            for optout in orm.Optout.objects.all():
+                try:
+                    user = orm['auth.User'].objects.get(email=optout.email)
+                    optout.user = user
+                    optout.save()
+                except ObjectDoesNotExist:
+                    # if user is not found (because they have already changed their email)
+                    # then delete the optout, as it's no longer useful.
+                    optout.delete()
+
+    def backwards(self, orm):
+
+        # backwards data migration to copy over emails of students to old email slot
+        if not db.dry_run:
+            for optout in orm.Optout.objects.all():
+                if optout.user is not None:
+                    optout.email = optout.user.email
+                    optout.save()
+
+    models = {
+        'auth.group': {
+            'Meta': {'object_name': 'Group'},
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'name': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '80'}),
+            'permissions': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Permission']", 'symmetrical': 'False', 'blank': 'True'})
+        },
+        'auth.permission': {
+            'Meta': {'ordering': "('content_type__app_label', 'content_type__model', 'codename')", 'unique_together': "(('content_type', 'codename'),)", 'object_name': 'Permission'},
+            'codename': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
+            'content_type': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['contenttypes.ContentType']"}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'name': ('django.db.models.fields.CharField', [], {'max_length': '50'})
+        },
+        'auth.user': {
+            'Meta': {'object_name': 'User'},
+            'date_joined': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime.now'}),
+            'email': ('django.db.models.fields.EmailField', [], {'max_length': '75', 'blank': 'True'}),
+            'first_name': ('django.db.models.fields.CharField', [], {'max_length': '30', 'blank': 'True'}),
+            'groups': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Group']", 'symmetrical': 'False', 'blank': 'True'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'is_active': ('django.db.models.fields.BooleanField', [], {'default': 'True'}),
+            'is_staff': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
+            'is_superuser': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
+            'last_login': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime.now'}),
+            'last_name': ('django.db.models.fields.CharField', [], {'max_length': '30', 'blank': 'True'}),
+            'password': ('django.db.models.fields.CharField', [], {'max_length': '128'}),
+            'user_permissions': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Permission']", 'symmetrical': 'False', 'blank': 'True'}),
+            'username': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '30'})
+        },
+        'bulk_email.courseemail': {
+            'Meta': {'object_name': 'CourseEmail'},
+            'course_id': ('django.db.models.fields.CharField', [], {'max_length': '255', 'db_index': 'True'}),
+            'created': ('django.db.models.fields.DateTimeField', [], {'auto_now_add': 'True', 'blank': 'True'}),
+            'html_message': ('django.db.models.fields.TextField', [], {'null': 'True', 'blank': 'True'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'modified': ('django.db.models.fields.DateTimeField', [], {'auto_now': 'True', 'blank': 'True'}),
+            'sender': ('django.db.models.fields.related.ForeignKey', [], {'default': '1', 'to': "orm['auth.User']", 'null': 'True', 'blank': 'True'}),
+            'slug': ('django.db.models.fields.CharField', [], {'max_length': '128', 'db_index': 'True'}),
+            'subject': ('django.db.models.fields.CharField', [], {'max_length': '128', 'blank': 'True'}),
+            'text_message': ('django.db.models.fields.TextField', [], {'null': 'True', 'blank': 'True'}),
+            'to_option': ('django.db.models.fields.CharField', [], {'default': "'myself'", 'max_length': '64'})
+        },
+        'bulk_email.optout': {
+            'Meta': {'unique_together': "(('user', 'course_id'),)", 'object_name': 'Optout'},
+            'course_id': ('django.db.models.fields.CharField', [], {'max_length': '255', 'db_index': 'True'}),
+            'email': ('django.db.models.fields.CharField', [], {'max_length': '255', 'db_index': 'True'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'user': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['auth.User']", 'null': 'True'})
+        },
+        'contenttypes.contenttype': {
+            'Meta': {'ordering': "('name',)", 'unique_together': "(('app_label', 'model'),)", 'object_name': 'ContentType', 'db_table': "'django_content_type'"},
+            'app_label': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'model': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
+            'name': ('django.db.models.fields.CharField', [], {'max_length': '100'})
+        }
+    }
+
+    complete_apps = ['bulk_email']
diff --git a/lms/djangoapps/bulk_email/migrations/0005_remove_optout_email.py b/lms/djangoapps/bulk_email/migrations/0005_remove_optout_email.py
new file mode 100644
index 0000000000..3639d1e473
--- /dev/null
+++ b/lms/djangoapps/bulk_email/migrations/0005_remove_optout_email.py
@@ -0,0 +1,78 @@
+# -*- coding: utf-8 -*-
+from south.db import db
+from south.v2 import SchemaMigration
+
+
+class Migration(SchemaMigration):
+
+    def forwards(self, orm):
+
+        # Deleting field 'Optout.email'
+        db.delete_column('bulk_email_optout', 'email')
+
+    def backwards(self, orm):
+
+        # Adding field 'Optout.email'
+        db.add_column('bulk_email_optout', 'email',
+                      self.gf('django.db.models.fields.CharField')(max_length=255, null=True, blank=True),
+                      keep_default=False)
+
+    models = {
+        'auth.group': {
+            'Meta': {'object_name': 'Group'},
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'name': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '80'}),
+            'permissions': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Permission']", 'symmetrical': 'False', 'blank': 'True'})
+        },
+        'auth.permission': {
+            'Meta': {'ordering': "('content_type__app_label', 'content_type__model', 'codename')", 'unique_together': "(('content_type', 'codename'),)", 'object_name': 'Permission'},
+            'codename': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
+            'content_type': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['contenttypes.ContentType']"}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'name': ('django.db.models.fields.CharField', [], {'max_length': '50'})
+        },
+        'auth.user': {
+            'Meta': {'object_name': 'User'},
+            'date_joined': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime.now'}),
+            'email': ('django.db.models.fields.EmailField', [], {'max_length': '75', 'blank': 'True'}),
+            'first_name': ('django.db.models.fields.CharField', [], {'max_length': '30', 'blank': 'True'}),
+            'groups': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Group']", 'symmetrical': 'False', 'blank': 'True'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'is_active': ('django.db.models.fields.BooleanField', [], {'default': 'True'}),
+            'is_staff': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
+            'is_superuser': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
+            'last_login': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime.now'}),
+            'last_name': ('django.db.models.fields.CharField', [], {'max_length': '30', 'blank': 'True'}),
+            'password': ('django.db.models.fields.CharField', [], {'max_length': '128'}),
+            'user_permissions': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Permission']", 'symmetrical': 'False', 'blank': 'True'}),
+            'username': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '30'})
+        },
+        'bulk_email.courseemail': {
+            'Meta': {'object_name': 'CourseEmail'},
+            'course_id': ('django.db.models.fields.CharField', [], {'max_length': '255', 'db_index': 'True'}),
+            'created': ('django.db.models.fields.DateTimeField', [], {'auto_now_add': 'True', 'blank': 'True'}),
+            'html_message': ('django.db.models.fields.TextField', [], {'null': 'True', 'blank': 'True'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'modified': ('django.db.models.fields.DateTimeField', [], {'auto_now': 'True', 'blank': 'True'}),
+            'sender': ('django.db.models.fields.related.ForeignKey', [], {'default': '1', 'to': "orm['auth.User']", 'null': 'True', 'blank': 'True'}),
+            'slug': ('django.db.models.fields.CharField', [], {'max_length': '128', 'db_index': 'True'}),
+            'subject': ('django.db.models.fields.CharField', [], {'max_length': '128', 'blank': 'True'}),
+            'text_message': ('django.db.models.fields.TextField', [], {'null': 'True', 'blank': 'True'}),
+            'to_option': ('django.db.models.fields.CharField', [], {'default': "'myself'", 'max_length': '64'})
+        },
+        'bulk_email.optout': {
+            'Meta': {'unique_together': "(('user', 'course_id'),)", 'object_name': 'Optout'},
+            'course_id': ('django.db.models.fields.CharField', [], {'max_length': '255', 'db_index': 'True'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'user': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['auth.User']", 'null': 'True'})
+        },
+        'contenttypes.contenttype': {
+            'Meta': {'ordering': "('name',)", 'unique_together': "(('app_label', 'model'),)", 'object_name': 'ContentType', 'db_table': "'django_content_type'"},
+            'app_label': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'model': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
+            'name': ('django.db.models.fields.CharField', [], {'max_length': '100'})
+        }
+    }
+
+    complete_apps = ['bulk_email']
diff --git a/lms/djangoapps/bulk_email/models.py b/lms/djangoapps/bulk_email/models.py
index 8fa864b955..72c9569cc1 100644
--- a/lms/djangoapps/bulk_email/models.py
+++ b/lms/djangoapps/bulk_email/models.py
@@ -30,7 +30,7 @@ class Email(models.Model):
     created = models.DateTimeField(auto_now_add=True)
     modified = models.DateTimeField(auto_now=True)
 
-    class Meta:
+    class Meta:  # pylint: disable=C0111
         abstract = True
 
 
@@ -61,10 +61,10 @@ class CourseEmail(Email, models.Model):
 
 class Optout(models.Model):
     """
-    Stores emails that have opted out of receiving emails from a course.
+    Stores users that have opted out of receiving emails from a course.
     """
-    email = models.CharField(max_length=255, db_index=True)
+    user = models.ForeignKey(User, db_index=True, null=True)
     course_id = models.CharField(max_length=255, db_index=True)
 
-    class Meta:
-        unique_together = ('email', 'course_id')
+    class Meta:  # pylint: disable=C0111
+        unique_together = ('user', 'course_id')
diff --git a/lms/djangoapps/bulk_email/tasks.py b/lms/djangoapps/bulk_email/tasks.py
index 35c57b9dad..7afd286570 100644
--- a/lms/djangoapps/bulk_email/tasks.py
+++ b/lms/djangoapps/bulk_email/tasks.py
@@ -2,10 +2,10 @@
 This module contains celery task functions for handling the sending of bulk email
 to a course.
 """
-import logging
 import math
 import re
 import time
+import gc
 
 from smtplib import SMTPServerDisconnected, SMTPDataError, SMTPConnectError
 
@@ -14,13 +14,14 @@ from django.contrib.auth.models import User, Group
 from django.core.mail import EmailMultiAlternatives, get_connection
 from django.http import Http404
 from celery import task, current_task
+from celery.utils.log import get_task_logger
 
 from bulk_email.models import CourseEmail, Optout
 from courseware.access import _course_staff_group_name, _course_instructor_group_name
 from courseware.courses import get_course_by_id
 from mitxmako.shortcuts import render_to_string
 
-log = logging.getLogger(__name__)
+log = get_task_logger(__name__)
 
 
 @task(default_retry_delay=10, max_retries=5)  # pylint: disable=E1102
@@ -47,37 +48,42 @@ def delegate_email_batches(email_id, to_option, course_id, course_url, user_id):
         raise delegate_email_batches.retry(arg=[email_id, to_option, course_id, course_url, user_id], exc=exc)
 
     if to_option == "myself":
-        recipient_qset = User.objects.filter(id=user_id).values('profile__name', 'email')
-
+        recipient_qset = User.objects.filter(id=user_id)
     elif to_option == "all" or to_option == "staff":
         staff_grpname = _course_staff_group_name(course.location)
         staff_group, _ = Group.objects.get_or_create(name=staff_grpname)
-        staff_qset = staff_group.user_set.values('profile__name', 'email')
+        staff_qset = staff_group.user_set.all()
         instructor_grpname = _course_instructor_group_name(course.location)
         instructor_group, _ = Group.objects.get_or_create(name=instructor_grpname)
-        instructor_qset = instructor_group.user_set.values('profile__name', 'email')
+        instructor_qset = instructor_group.user_set.all()
         recipient_qset = staff_qset | instructor_qset
 
         if to_option == "all":
-            # Two queries are executed per performance considerations for MySQL.
-            # See https://docs.djangoproject.com/en/1.2/ref/models/querysets/#in.
-            course_optouts = Optout.objects.filter(course_id=course_id).values_list('email', flat=True)
-            enrollment_qset = User.objects.filter(courseenrollment__course_id=course_id).exclude(email__in=list(course_optouts)).values('profile__name', 'email')
+            enrollment_qset = User.objects.filter(courseenrollment__course_id=course_id,
+                                                  courseenrollment__is_active=True)
             recipient_qset = recipient_qset | enrollment_qset
         recipient_qset = recipient_qset.distinct()
-
     else:
         log.error("Unexpected bulk email TO_OPTION found: %s", to_option)
         raise Exception("Unexpected bulk email TO_OPTION found: {0}".format(to_option))
 
-    recipient_list = list(recipient_qset)
-    total_num_emails = len(recipient_list)
-    num_workers = int(math.ceil(float(total_num_emails) / float(settings.EMAILS_PER_TASK)))
-    chunk = int(math.ceil(float(total_num_emails) / float(num_workers)))
-
-    for i in range(num_workers):
-        to_list = recipient_list[i * chunk:i * chunk + chunk]
-        course_email.delay(email_id, to_list, course.display_name, course_url, False)
+    recipient_qset = recipient_qset.order_by('pk')
+    total_num_emails = recipient_qset.count()
+    num_queries = int(math.ceil(float(total_num_emails) / float(settings.EMAILS_PER_QUERY)))
+    last_pk = recipient_qset[0].pk - 1
+    num_workers = 0
+    for j in range(num_queries):
+        recipient_sublist = list(recipient_qset.order_by('pk').filter(pk__gt=last_pk)
+                                 .values('profile__name', 'email', 'pk')[:settings.EMAILS_PER_QUERY])
+        last_pk = recipient_sublist[-1]['pk']
+        num_emails_this_query = len(recipient_sublist)
+        num_tasks_this_query = int(math.ceil(float(num_emails_this_query) / float(settings.EMAILS_PER_TASK)))
+        chunk = int(math.ceil(float(num_emails_this_query) / float(num_tasks_this_query)))
+        for i in range(num_tasks_this_query):
+            to_list = recipient_sublist[i * chunk:i * chunk + chunk]
+            course_email.delay(email_id, to_list, course.display_name, course_url, False)
+        num_workers += num_tasks_this_query
+        gc.collect()
     return num_workers
 
 
@@ -89,12 +95,22 @@ def course_email(email_id, to_list, course_title, course_url, throttle=False):
     being the only "to".  Emails are sent multipart, in both plain
     text and html.
     """
+
     try:
         msg = CourseEmail.objects.get(id=email_id)
     except CourseEmail.DoesNotExist as exc:
         log.exception(exc.args[0])
         raise exc
 
+    # exclude optouts
+    optouts = Optout.objects.filter(course_id=msg.course_id,
+                                    user__email__in=[i['email'] for i in to_list])\
+                            .values_list('user__email', flat=True)
+
+    num_optout = len(optouts)
+
+    to_list = filter(lambda x: x['email'] not in optouts, to_list)
+
     subject = "[" + course_title + "] " + msg.subject
 
     course_title_no_quotes = re.sub(r'"', '', course_title)
@@ -114,9 +130,9 @@ def course_email(email_id, to_list, course_title, course_url, throttle=False):
         }
 
         while to_list:
-            (name, email) = to_list[-1].values()
-            email_context['name'] = name
+            email = to_list[-1]['email']
             email_context['email'] = email
+            email_context['name'] = to_list[-1]['profile__name']
 
             html_footer = render_to_string(
                 'emails/email_footer.html',
@@ -157,7 +173,7 @@ def course_email(email_id, to_list, course_title, course_url, throttle=False):
             to_list.pop()
 
         connection.close()
-        return course_email_result(num_sent, num_error)
+        return course_email_result(num_sent, num_error, num_optout)
 
     except (SMTPDataError, SMTPConnectError, SMTPServerDisconnected) as exc:
         # Error caught here cause the email to be retried.  The entire task is actually retried without popping the list
@@ -175,6 +191,6 @@ def course_email(email_id, to_list, course_title, course_url, throttle=False):
 
 
 # This string format code is wrapped in this function to allow mocking for a unit test
-def course_email_result(num_sent, num_error):
+def course_email_result(num_sent, num_error, num_optout):
     """Return the formatted result of course_email sending."""
-    return "Sent {0}, Fail {1}".format(num_sent, num_error)
+    return "Sent {0}, Fail {1}, Optout {2}".format(num_sent, num_error, num_optout)
diff --git a/lms/djangoapps/bulk_email/tests/test_course_optout.py b/lms/djangoapps/bulk_email/tests/test_course_optout.py
index 499ccb0b95..18f04cebc1 100644
--- a/lms/djangoapps/bulk_email/tests/test_course_optout.py
+++ b/lms/djangoapps/bulk_email/tests/test_course_optout.py
@@ -10,6 +10,7 @@ from django.test.utils import override_settings
 
 from courseware.tests.tests import TEST_DATA_MONGO_MODULESTORE
 from student.tests.factories import UserFactory, AdminFactory, CourseEnrollmentFactory
+from student.models import CourseEnrollment
 from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
 from xmodule.modulestore.tests.factories import CourseFactory
 
@@ -94,6 +95,8 @@ class TestOptoutCourseEmails(ModuleStoreTestCase):
 
         self.client.logout()
 
+        self.assertTrue(CourseEnrollment.is_enrolled(self.student, self.course.id))
+
         self.client.login(username=self.instructor.username, password="test")
         self.navigate_to_email_view()
 
diff --git a/lms/djangoapps/bulk_email/tests/test_email.py b/lms/djangoapps/bulk_email/tests/test_email.py
index 67f49d1f51..b0cf8dc06e 100644
--- a/lms/djangoapps/bulk_email/tests/test_email.py
+++ b/lms/djangoapps/bulk_email/tests/test_email.py
@@ -2,7 +2,6 @@
 """
 Unit tests for sending course email
 """
-
 from django.test.utils import override_settings
 from django.conf import settings
 from django.core import mail
@@ -14,12 +13,29 @@ from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
 from xmodule.modulestore.tests.factories import CourseFactory
 
 from bulk_email.tasks import delegate_email_batches, course_email
-from bulk_email.models import CourseEmail
+from bulk_email.models import CourseEmail, Optout
 
 from mock import patch
 
 STAFF_COUNT = 3
 STUDENT_COUNT = 10
+LARGE_NUM_EMAILS = 137
+
+
+class MockCourseEmailResult(object):
+    """
+    A small closure-like class to keep count of emails sent over all tasks, recorded
+    by mock object side effects
+    """
+    emails_sent = 0
+
+    def get_mock_course_email_result(self):
+        """Wrapper for mock email function."""
+        def mock_course_email_result(sent, failed, output, **kwargs):  # pylint: disable=W0613
+            """Increments count of number of emails sent."""
+            self.emails_sent += sent
+            return True
+        return mock_course_email_result
 
 
 @override_settings(MODULESTORE=TEST_DATA_MONGO_MODULESTORE)
@@ -110,6 +126,7 @@ class TestEmailSendFromDashboard(ModuleStoreTestCase):
 
         self.assertContains(response, "Your email was successfully queued for sending.")
 
+        # the 1 is for the instructor in this test and others
         self.assertEquals(len(mail.outbox), 1 + len(self.staff))
         self.assertItemsEqual(
             [e.to[0] for e in mail.outbox],
@@ -225,6 +242,43 @@ class TestEmailSendFromDashboard(ModuleStoreTestCase):
             [self.instructor.email] + [s.email for s in self.staff] + [s.email for s in self.students]
         )
 
+    @override_settings(EMAILS_PER_TASK=3, EMAILS_PER_QUERY=7)
+    @patch('bulk_email.tasks.course_email_result')
+    def test_chunked_queries_send_numerous_emails(self, email_mock):
+        """
+        Test sending a large number of emails, to test the chunked querying
+        """
+        mock_factory = MockCourseEmailResult()
+        email_mock.side_effect = mock_factory.get_mock_course_email_result()
+        added_users = []
+        for _ in xrange(LARGE_NUM_EMAILS):
+            user = UserFactory()
+            added_users.append(user)
+            CourseEnrollmentFactory.create(user=user, course_id=self.course.id)
+
+        optouts = []
+        for i in [1, 3, 9, 10, 18]:  # 5 random optouts
+            user = added_users[i]
+            optouts.append(user)
+            optout = Optout(user=user, course_id=self.course.id)
+            optout.save()
+
+        test_email = {
+            'action': 'Send email',
+            'to_option': 'all',
+            'subject': 'test subject for all',
+            'message': 'test message for all'
+        }
+        response = self.client.post(self.url, test_email)
+        self.assertContains(response, "Your email was successfully queued for sending.")
+        self.assertEquals(mock_factory.emails_sent,
+                          1 + len(self.staff) + len(self.students) + LARGE_NUM_EMAILS - len(optouts))
+        self.assertItemsEqual(
+            [e.to[0] for e in mail.outbox],
+            [self.instructor.email] + [s.email for s in self.staff] + [s.email for s in self.students] +
+            [s.email for s in added_users if s not in optouts]
+        )
+
 
 @override_settings(MODULESTORE=TEST_DATA_MONGO_MODULESTORE)
 class TestEmailSendExceptions(ModuleStoreTestCase):
diff --git a/lms/djangoapps/bulk_email/tests/test_err_handling.py b/lms/djangoapps/bulk_email/tests/test_err_handling.py
index e7d2e62d4a..606a0bef88 100644
--- a/lms/djangoapps/bulk_email/tests/test_err_handling.py
+++ b/lms/djangoapps/bulk_email/tests/test_err_handling.py
@@ -94,7 +94,8 @@ class TestEmailErrors(ModuleStoreTestCase):
         # We shouldn't retry when hitting a 5xx error
         self.assertFalse(retry.called)
         # Test that after the rejected email, the rest still successfully send
-        ((sent, fail), _) = result.call_args
+        ((sent, fail, optouts), _) = result.call_args
+        self.assertEquals(optouts, 0)
         self.assertEquals(fail, 1)
         self.assertEquals(sent, settings.EMAILS_PER_TASK - 1)
 
diff --git a/lms/djangoapps/instructor/views/legacy.py b/lms/djangoapps/instructor/views/legacy.py
index a2ccf0a5bc..15b0df59c1 100644
--- a/lms/djangoapps/instructor/views/legacy.py
+++ b/lms/djangoapps/instructor/views/legacy.py
@@ -56,7 +56,6 @@ from mitxmako.shortcuts import render_to_string
 
 from bulk_email.models import CourseEmail
 from html_to_text import html_to_text
-import datetime
 from bulk_email import tasks
 
 log = logging.getLogger(__name__)
@@ -66,11 +65,11 @@ FORUM_ROLE_ADD = 'add'
 FORUM_ROLE_REMOVE = 'remove'
 
 
-def split_by_comma_and_whitespace(s):
+def split_by_comma_and_whitespace(a_str):
     """
-    Return string s, split by , or whitespace
+    Return string a_str, split by , or whitespace
     """
-    return re.split(r'[\s,]', s)
+    return re.split(r'[\s,]', a_str)
 
 
 @ensure_csrf_cookie
@@ -124,13 +123,13 @@ def instructor_dashboard(request, course_id):
         datatable['data'] = data
         return datatable
 
-    def return_csv(fn, datatable, fp=None):
+    def return_csv(func, datatable, file_pointer=None):
         """Outputs a CSV file from the contents of a datatable."""
-        if fp is None:
+        if file_pointer is None:
             response = HttpResponse(mimetype='text/csv')
-            response['Content-Disposition'] = 'attachment; filename={0}'.format(fn)
+            response['Content-Disposition'] = 'attachment; filename={0}'.format(func)
         else:
-            response = fp
+            response = file_pointer
         writer = csv.writer(response, dialect='excel', quotechar='"', quoting=csv.QUOTE_ALL)
         writer.writerow(datatable['header'])
         for datarow in datatable['data']:
@@ -279,11 +278,11 @@ def instructor_dashboard(request, course_id):
                 msg += '<font color="red">Failed to create a background task for rescoring "{0}".</font>'.format(problem_url)
             else:
                 track.views.server_track(request, "rescore-all-submissions", {"problem": problem_url, "course": course_id}, page="idashboard")
-        except ItemNotFoundError as e:
+        except ItemNotFoundError as err:
             msg += '<font color="red">Failed to create a background task for rescoring "{0}": problem not found.</font>'.format(problem_url)
-        except Exception as e:
-            log.error("Encountered exception from rescore: {0}".format(e))
-            msg += '<font color="red">Failed to create a background task for rescoring "{0}": {1}.</font>'.format(problem_url, e.message)
+        except Exception as err:
+            log.error("Encountered exception from rescore: {0}".format(err))
+            msg += '<font color="red">Failed to create a background task for rescoring "{0}": {1}.</font>'.format(problem_url, err.message)
 
     elif "Reset ALL students' attempts" in action:
         problem_urlname = request.POST.get('problem_for_all_students', '')
@@ -294,12 +293,12 @@ def instructor_dashboard(request, course_id):
                 msg += '<font color="red">Failed to create a background task for resetting "{0}".</font>'.format(problem_url)
             else:
                 track.views.server_track(request, "reset-all-attempts", {"problem": problem_url, "course": course_id}, page="idashboard")
-        except ItemNotFoundError as e:
-            log.error('Failure to reset: unknown problem "{0}"'.format(e))
+        except ItemNotFoundError as err:
+            log.error('Failure to reset: unknown problem "{0}"'.format(err))
             msg += '<font color="red">Failed to create a background task for resetting "{0}": problem not found.</font>'.format(problem_url)
-        except Exception as e:
-            log.error("Encountered exception from reset: {0}".format(e))
-            msg += '<font color="red">Failed to create a background task for resetting "{0}": {1}.</font>'.format(problem_url, e.message)
+        except Exception as err:
+            log.error("Encountered exception from reset: {0}".format(err))
+            msg += '<font color="red">Failed to create a background task for resetting "{0}": {1}.</font>'.format(problem_url, err.message)
 
     elif "Show Background Task History for Student" in action:
         # put this before the non-student case, since the use of "in" will cause this to be missed
@@ -475,10 +474,10 @@ def instructor_dashboard(request, course_id):
                     return return_csv('grades %s.csv' % aname, datatable)
 
                 elif 'remote gradebook' in action:
-                    fp = StringIO()
-                    return_csv('', datatable, fp=fp)
-                    fp.seek(0)
-                    files = {'datafile': fp}
+                    file_pointer = StringIO()
+                    return_csv('', datatable, file_pointer=file_pointer)
+                    file_pointer.seek(0)
+                    files = {'datafile': file_pointer}
                     msg2, _ = _do_remote_gradebook(request.user, course, 'post-grades', files=files)
                     msg += msg2
 
diff --git a/lms/envs/aws.py b/lms/envs/aws.py
index 45738e6d31..a22fbc5bb6 100644
--- a/lms/envs/aws.py
+++ b/lms/envs/aws.py
@@ -106,7 +106,7 @@ EMAIL_HOST = ENV_TOKENS.get('EMAIL_HOST', 'localhost')  # django default is loca
 EMAIL_PORT = ENV_TOKENS.get('EMAIL_PORT', 25)  # django default is 25
 EMAIL_USE_TLS = ENV_TOKENS.get('EMAIL_USE_TLS', False)  # django default is False
 EMAILS_PER_TASK = ENV_TOKENS.get('EMAILS_PER_TASK', 100)
-
+EMAILS_PER_QUERY = ENV_TOKENS.get('EMAILS_PER_QUERY', 1000)
 SITE_NAME = ENV_TOKENS['SITE_NAME']
 SESSION_ENGINE = ENV_TOKENS.get('SESSION_ENGINE', SESSION_ENGINE)
 SESSION_COOKIE_DOMAIN = ENV_TOKENS.get('SESSION_COOKIE_DOMAIN')
diff --git a/lms/envs/common.py b/lms/envs/common.py
index 51fba25e69..5ad81d5f03 100644
--- a/lms/envs/common.py
+++ b/lms/envs/common.py
@@ -364,7 +364,8 @@ IGNORABLE_404_ENDS = ('favicon.ico')
 EMAIL_BACKEND = 'django.core.mail.backends.console.EmailBackend'
 DEFAULT_FROM_EMAIL = 'registration@edx.org'
 DEFAULT_BULK_FROM_EMAIL = 'course-updates@edx.org'
-EMAILS_PER_TASK = 10
+EMAILS_PER_TASK = 100
+EMAILS_PER_QUERY = 1000
 DEFAULT_FEEDBACK_EMAIL = 'feedback@edx.org'
 SERVER_EMAIL = 'devops@edx.org'
 TECH_SUPPORT_EMAIL = 'technical@edx.org'